package com.atguigu.table

import com.atguigu.api.SensorReading
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.scala._

/**
 * @ClassName Example
 * @Description table api和 table sql
 * @Author Mr Yang
 * @Date 2020/10/4 12:27
 * @Version 1.0
 */
object Example {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val source = getClass.getResource("/sensor.txt")
    val dataStream = env.readTextFile(source.getPath)
      .map( data =>{
        val dataArray = data.split(",")
        SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
      })

    //创建表环境
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)
    //将流转换成表
    val dataTable: Table = tableEnv.fromDataStream(dataStream)
    //调用table api，做转换操作(需要隐士转换)
    val resultTable: Table = dataTable
      .select("id,temperature")
      .filter("id == 'sensor_1'")
    //flink sql转换方式
    tableEnv.createTemporaryView("dataTable",dataTable)
    val resultSqlTable: Table = tableEnv.sqlQuery(
      """
        |select id, temperature from dataTable where id = 'sensor_1'
        |""".stripMargin)

    //把表转换成流
    val tableStream: DataStream[(String, Double)] = resultSqlTable.toAppendStream[(String, Double)]

    tableStream.print()
    env.execute("table api job")
  }
}
